use crate::{backup::{SeqChunk, SeqFileInfo}, pipeline::MutExecutor};
use super::*;
use crate::common::chunking::*;
use tokio::time::Instant;

#[derive(Clone, Debug)]
pub struct SplitConfig {
    pub continuous_split:bool,
    pub chunking:ChunkingSelector,
}

impl SplitConfig {
    pub fn from_source(source:&SourcePool) -> Option<Self> {
        let continuous_split= source.get_config().get("continuous_split").ok()?;
        let chunking = source.chunking.clone();
        Some(Self { chunking, continuous_split })
    }

    pub fn build(self) -> BufferSplitor {
        BufferSplitor::new(self.chunking, self.continuous_split)
    }

    pub fn build_file_splitor(&self, file_id:usize, name:String) -> FileSplitor {
        FileSplitor::new(file_id, name).continuous(self.continuous_split)
    }
}

pub struct FileSplitor {
    continuous:bool,
    file_id:usize,
    name:String,
    buf_id:usize,
    chunk_id:usize,
    buf:BytesMut,
}

impl FileSplitor {
    pub fn new(file_id:usize, name:String) -> Self {
        Self { 
            continuous:false,
            file_id, 
            name,
            buf_id: 1, 
            chunk_id: 0,
            buf: BytesMut::new(),
        }
    }

    pub fn alloc_chunk_id(&mut self) -> usize {
        self.chunk_id += 1;
        self.chunk_id - 1
    }

    pub fn commit_chunk<F:FnMut(WriteItem)>(&mut self, chunk:Chunk, produce_chunk:&mut F) {
        let seq_chunk = SeqChunk { 
            file_id: self.file_id, 
            chunk_id: self.alloc_chunk_id(), 
            chunk,
        };
        produce_chunk(WriteItem::Chunk(seq_chunk));
    }

    pub fn continuous(mut self, val:bool) -> Self {
        self.continuous = val;
        self
    }

    pub fn recv_end<F:FnMut(WriteItem)>(&mut self, produce_chunk:&mut F) {
        let mut swap_buf = BytesMut::new();
        std::mem::swap(&mut self.buf, &mut swap_buf);
        if swap_buf.len() != 0 {
            let chunk = Chunk::from_bytes_mut(swap_buf);
            self.commit_chunk(chunk, produce_chunk);
        }
        let file_info = FileInfo { 
            file_name: self.name.clone(), 
            block_num: self.chunk_id,
        };
        let seq_file_info = SeqFileInfo {
            file_id:self.file_id,
            file_info:file_info,
        };
        produce_chunk(WriteItem::File(seq_file_info));
    }

    pub fn recv_read_buf<F:FnMut(WriteItem)>(&mut self, buf_id:usize, mut buf:BytesMut, produce_chunk:&mut F) {
        if self.buf_id == buf_id {
            self.buf.unsplit(buf);
            return;
        } else if self.continuous && self.buf.len() != 0 {
            panic!("continuous is true, but last buffer is existed");
        }
        if self.buf.len() != 0 {
            std::mem::swap(&mut self.buf, &mut buf);
            let chunk = Chunk::from_bytes_mut(buf);
            self.commit_chunk(chunk, produce_chunk);
        } else {
            self.buf_id = buf_id;
            self.buf = buf;
        }
    }

    pub fn recv_tmp_buf(&mut self, buf_id:usize, mut buf:BytesMut) {
        if self.buf_id == buf_id {
            panic!("tmp buffer id = current buf id");
        } else if !self.continuous {
            panic!("continous is false, but receive tmp buffer");
        }
        if self.buf.len() < buf.len() {
            let mut new_buf = buf.split_off(buf.len() - self.buf.len());
            new_buf.copy_from_slice(&self.buf[..]);
            self.buf_id = buf_id;
            self.buf = new_buf;
        } else {
            panic!("current buf len > tmp buf len")
        }
    }

    pub fn do_split<F:FnMut(WriteItem)>(
        &mut self, 
        chunking_algorithm:&ChunkingSelector,
        produce_write_item:&mut F,
    ) {
        while chunking_algorithm.min_len() < self.buf.len() {
            let chunk_size = chunking_algorithm.split(&self.buf);
            let chunk_bytes = self.buf.split_to(chunk_size);
            let chunk = Chunk::from_bytes_mut(chunk_bytes);
            self.commit_chunk(chunk, produce_write_item);
        }
    }

    pub fn recv_and_split<F:FnMut(WriteItem)>(
        &mut self,
        buf_id:usize, 
        buf:BytesMut,
        chunking_algorithm:&ChunkingSelector,
        produce_write_item:&mut F,
    ) {
        self.recv_read_buf(buf_id, buf, produce_write_item);
        self.do_split(chunking_algorithm, produce_write_item);
    }

    pub fn recv_file_block<F:FnMut(WriteItem)>(
        &mut self, 
        file_block:FileBlock, 
        chunking_algorithm:&ChunkingSelector,
        produce_write_item:&mut F
    ) {
        if file_block.id != self.file_id {
            panic!("file id not match");
        }
        match file_block.kind {
            FileKind::Name(_) => panic!("file splitor receive name block"),
            FileKind::Data(id, buf) => self.recv_and_split(id, buf, chunking_algorithm, produce_write_item),
            FileKind::Tmp(id, buf) => self.recv_tmp_buf(id, buf),
            FileKind::FileEnd => self.recv_end(produce_write_item),
        }
    }
}

pub struct BufferSplitor {
    continous_split:bool,
    chunking_algorithm:ChunkingSelector,
    current_file_splitor:Option<FileSplitor>,
    split_time:u128,
}

impl BufferSplitor {
    pub fn new(chunking:ChunkingSelector, continuous:bool) -> Self {
        Self { 
            continous_split:continuous,
            chunking_algorithm:chunking,
            current_file_splitor:None,
            split_time:0,
        }
    }

    fn build_file_splitor(&mut self, file_id:usize, file_name:String) {
        self.current_file_splitor = Some(FileSplitor::new(file_id, file_name).continuous(self.continous_split));
    }

    pub fn split_file_block<F:FnMut(WriteItem)>(&mut self, file_block:FileBlock, mut produce_write_item:F) -> BackupResult<()> {
        let file_id = file_block.id;
        if let FileKind::Name(file_name) = file_block.kind {
            self.build_file_splitor(file_id, file_name);
        } else {
            let start_time = Instant::now();
            self.current_file_splitor.as_mut().unwrap().recv_file_block(file_block, &self.chunking_algorithm, &mut produce_write_item);
            self.split_time += start_time.elapsed().as_micros();
        }
        Ok(())
    }

    pub fn get_split_time(&self) -> u128 {
        self.split_time
    }
}

impl Evaluated for BufferSplitor {
    fn register(&self, evaluation:&Evaluation) {
        let mut eva = evaluation.lock();
        eva.register(BackupEvaluation::Time, "split time(us)");
    }

    fn set_value(&self, evaluation:&Evaluation) {
        let mut eva = evaluation.lock();
        eva.set("split time(us)", self.get_split_time());
    }
}

impl MutExecutor<FileBlock, WriteItem> for BufferSplitor {
    type Error = BackupError;
    fn exec<'scope, F:FnMut(WriteItem)>(&'scope mut self, input:FileBlock, output:F) -> Result<(), Self::Error> {
        self.split_file_block(input, output)
    }
}

// use std::collections::HashMap;
// pub struct SplitorTable {
//     continous_split:bool,
//     table:HashMap<(usize, usize), std::sync::mpsc::Sender<FileBlock>>,
// }

// impl SplitorTable {
//     pub fn new(continous_split:bool) -> Self {
//         Self { continous_split, table: HashMap::new() }
//     }

//     pub fn get_sender(&mut self, key:(usize, usize)) -> std::sync::mpsc::Sender<FileBlock> {
//         // if let Some(sender) = self.table.get(&key) {

//         // } else {
//         //     let sender = 
//         //     self.table.insert(k, v)
//         // }
//         todo!()
//     }

//     pub fn forward_file_block(&self, file_block:FileBlock) {
//         // match file_block.kind {
//         //     FileKind::Name(_) => {},

//         // }
//     }
// }